package com.atguigu.stream.window


import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.lang

object AvgTempByAggAndAvg {
  case class AvgInfo (id :String,avgTemp :Double,windowStart:Long,winowEnd:Long)
  def main(args: Array[String]): Unit = {

    case class AvgInfo (id :String,avgTemp :Double,windowStart:Long,winowEnd:Long)

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream = env.addSource(new SensorSource)
    stream.keyBy(_.id)
      .timeWindow(Time.seconds(10))
      .aggregate(new AvgTemoAgg,new WindowResult)
      .print()

    env.execute()

  }


  class AvgTemoAgg extends AggregateFunction[SensorReading,(String,Long,Double),(String,Double)]{
    //    创建空累加器
    override def createAccumulator(): (String, Long, Double) =("",0L,0.0)
    //    聚合逻辑
    override def add(in: SensorReading, acc: (String, Long, Double)): (String, Long, Double) = {
      (in.id,acc._2+1,acc._3+in.temperature)

    }
    //    窗口聚合时输出什么
    override def getResult(acc: (String, Long, Double)): (String, Double) = {
      (acc._1,acc._3/acc._2)


    }
    //    累加器聚合逻辑是什么
    override def merge(acc: (String, Long, Double), acc1: (String, Long, Double)): (String, Long, Double) = {

      (acc._1,acc._2+acc1._2,acc._3+acc1._3)

    }






}
    class WindowResult extends ProcessWindowFunction[(String,Double),AvgInfo,String,TimeWindow]{

      //每个窗口关闭，只有一条数据

      override def process(key: String, context: Context, elements: Iterable[(String, Double)], out: Collector[AvgInfo]): Unit = {

        out.collect(AvgInfo(key,elements.head._2,context.window.getStart,context.window.getEnd))
      }
    }


}
